fix: use next_expected_byte() in retried uploads (#3037)

Author: dopiera

google/cloud/storage/internal/retry_resumable_upload_session.cc

@@ -38,38 +38,66 @@ StatusOr<ResumableUploadResponse> ReturnError(Status&& last_status,
 
 StatusOr<ResumableUploadResponse> RetryResumableUploadSession::UploadChunk(
     std::string const& buffer) {
-  Status last_status(StatusCode::kDeadlineExceeded,
-                     "Retry policy exhausted before first attempt was made.");
-  while (!retry_policy_->IsExhausted()) {
-    auto result = session_->UploadChunk(buffer);
-    if (result.ok()) {
-      return result;
-    }
-    last_status = std::move(result).status();
-    if (!retry_policy_->OnFailure(last_status)) {
-      return ReturnError(std::move(last_status), *retry_policy_, __func__);
-    }
-    auto delay = backoff_policy_->OnCompletion();
-    std::this_thread::sleep_for(delay);
-
-    result = ResetSession();
-    if (!result.ok()) {
-      return result;
-    }
-  }
-  std::ostringstream os;
-  os << "Retry policy exhausted in " << __func__ << ": " << last_status;
-  return Status(last_status.code(), os.str());
+  return UploadGenericChunk(buffer, optional<std::uint64_t>());
 }
 
 StatusOr<ResumableUploadResponse> RetryResumableUploadSession::UploadFinalChunk(
     std::string const& buffer, std::uint64_t upload_size) {
+  return UploadGenericChunk(buffer, upload_size);
+}
+
+StatusOr<ResumableUploadResponse>
+RetryResumableUploadSession::UploadGenericChunk(
+    std::string const& buffer, optional<std::uint64_t> const& upload_size) {
+  bool const is_final_chunk = upload_size.has_value();
+  char const* const func = is_final_chunk ? "UploadFinalChunk" : "UploadChunk";
+  std::uint64_t next_byte = session_->next_expected_byte();
   Status last_status(StatusCode::kDeadlineExceeded,
                      "Retry policy exhausted before first attempt was made.");
+  // On occasion, we might need to retry uploading only a part of the buffer.
+  // The current APIs require us to copy the buffer in such a scenario. We can
+  // and want to avoid the copy in the common case, so we use this variable to
+  // either reference the copy or the original buffer.
+  // TODO(#3036): change the APIs to avoid this extra copy.
+  std::string const* buffer_to_use = &buffer;
+  std::string truncated_buffer;
   while (!retry_policy_->IsExhausted()) {
-    auto result = session_->UploadFinalChunk(buffer, upload_size);
+    std::uint64_t new_next_byte = session_->next_expected_byte();
+    if (new_next_byte < next_byte) {
+      std::stringstream os;
+      os << func << ": server has not confirmed bytes which we no longer hold ("
+         << new_next_byte << "-" << next_byte
+         << "). This is most likely a bug in the GCS client. Please report it "
+            "at https://github.com/googleapis/google-cloud-cpp/issues/new";
+      return Status(StatusCode::kInternal, os.str());
+    }
+    if (new_next_byte > next_byte) {
+      truncated_buffer = buffer_to_use->substr(new_next_byte - next_byte);
+      buffer_to_use = &truncated_buffer;
+      next_byte = new_next_byte;
+    }
+    auto result = is_final_chunk
+                      ? session_->UploadFinalChunk(*buffer_to_use, *upload_size)
+                      : session_->UploadChunk(*buffer_to_use);
     if (result.ok()) {
-      return result;
+      if (is_final_chunk &&
+          result->upload_state == ResumableUploadResponse::kDone) {
+        // If it's a final chunk and it succeded, return.
+        return result;
+      }
+      if (next_expected_byte() - next_byte == buffer_to_use->size()) {
+        // Otherwise, return only if there were no failures and it wasn't a
+        // short write.
+        return result;
+      }
+      std::stringstream os;
+      os << "Short write. Previous next_byte=" << next_byte
+         << ", current next_byte=" << next_expected_byte()
+         << ", inteded to write " << buffer_to_use->size();
+      last_status = Status(StatusCode::kUnavailable, os.str());
+      // Don't reset the session on a short write nor wait according to the
+      // backoff policy - we did get a response from the server after all.
+      continue;
     }
     last_status = std::move(result).status();
     if (!retry_policy_->OnFailure(last_status)) {
@@ -84,7 +112,7 @@ StatusOr<ResumableUploadResponse> RetryResumableUploadSession::UploadFinalChunk(
     }
   }
   std::ostringstream os;
-  os << "Retry policy exhausted in " << __func__ << ": " << last_status;
+  os << "Retry policy exhausted in " << func << ": " << last_status;
   return Status(last_status.code(), os.str());
 }
 

google/cloud/storage/internal/retry_resumable_upload_session.h

@@ -52,6 +52,9 @@ class RetryResumableUploadSession : public ResumableUploadSession {
   StatusOr<ResumableUploadResponse> const& last_response() const override;
 
  private:
+  // Retry either UploadChunk or either UploadFinalChunk.
+  StatusOr<ResumableUploadResponse> UploadGenericChunk(
+      std::string const& buffer, optional<std::uint64_t> const& upload_size);
   std::unique_ptr<ResumableUploadSession> session_;
   std::unique_ptr<RetryPolicy> retry_policy_;
   std::unique_ptr<BackoffPolicy> backoff_policy_;