Non throwing object write stream (#1749)

Unless the application explicitly requests exceptions by setting the
exception mask. By default `std::ios_base::badbit` is set when an error
is detected, and the ObjectWriteStream saves the status with a more
detailed error message.

This change will reduce code coverage because many errors that used to
be propagated as exceptions now require an extra if(), but we do not
have test infrastructure to produce these error conditions on demand.

Wrote a test to verify that resumable upload errors are reported
correctly. Fixed the testbench to properly validate preconditions on a
resumable upload. Then found a bug in how we handle error status for
resumable upload failures.

Author: coryan

ci/test-install/storage_install_test.cc

@@ -32,17 +32,19 @@ int main(int argc, char* argv[]) try {
   gcs::Client client;
 
   gcs::ObjectWriteStream os = client.WriteObject(bucket_name, object_name);
+  os.exceptions(std::ios_base::badbit | std::ios_base::failbit);
   os << "Hello World" << std::endl;
-  gcs::ObjectMetadata meta = os.Close();
+  os.Close();
+  gcs::ObjectMetadata meta = os.metadata().value();
   std::cout << "Successfully created object, generation=" << meta.generation()
             << std::endl;
 
   gcs::ObjectReadStream stream = client.ReadObject(bucket_name, object_name);
+  stream.exceptions(std::ios_base::badbit | std::ios_base::failbit);
 
   int count = 0;
-  while (not stream.eof()) {
-    std::string line;
-    std::getline(stream, line, '\n');
+  std::string line;
+  while (std::getline(stream, line, '\n')) {
     ++count;
   }
   client.DeleteObject(bucket_name, object_name,

google/cloud/storage/client_write_object_test.cc

@@ -59,7 +59,7 @@ class WriteObjectTest : public ::testing::Test {
 class MockStreambuf : public internal::ObjectWriteStreambuf {
  public:
   MOCK_CONST_METHOD0(IsOpen, bool());
-  MOCK_METHOD0(DoClose, internal::HttpResponse());
+  MOCK_METHOD0(DoClose, StatusOr<internal::HttpResponse>());
   MOCK_METHOD1(ValidateHash, void(ObjectMetadata const&));
   MOCK_CONST_METHOD0(received_hash, std::string const&());
   MOCK_CONST_METHOD0(computed_hash, std::string const&());
@@ -88,7 +88,8 @@ TEST_F(WriteObjectTest, WriteObject) {
 
   auto stream = client->WriteObject("test-bucket-name", "test-object-name");
   stream << "Hello World!";
-  ObjectMetadata actual = stream.Close();
+  stream.Close();
+  ObjectMetadata actual = stream.metadata().value();
   EXPECT_EQ(expected, actual);
 }
 

google/cloud/storage/examples/storage_bucket_samples.cc

@@ -477,7 +477,8 @@ void WriteObjectRequesterPays(google::cloud::storage::Client client, int& argc,
       stream << (lineno + 1) << ": I will write better examples\n";
     }
 
-    gcs::ObjectMetadata meta = stream.Close();
+    stream.Close();
+    gcs::ObjectMetadata meta = stream.metadata().value();
     std::cout << "The resulting object size is: " << meta.size() << std::endl;
   }
   //! [write object requester pays]

google/cloud/storage/examples/storage_object_samples.cc

@@ -279,7 +279,8 @@ void WriteObject(google::cloud::storage::Client client, int& argc,
       stream << (lineno + 1) << ": " << text << "\n";
     }
 
-    gcs::ObjectMetadata meta = stream.Close();
+    stream.Close();
+    gcs::ObjectMetadata meta = stream.metadata().value();
     std::cout << "The resulting object size is: " << meta.size() << std::endl;
   }
   //! [write object]
@@ -382,7 +383,8 @@ non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.
 )""";
     }
 
-    gcs::ObjectMetadata metadata = stream.Close();
+    stream.Close();
+    gcs::ObjectMetadata metadata = stream.metadata().value();
     std::cout << "Upload completed, the new object metadata is: " << metadata
               << std::endl;
   }
@@ -754,7 +756,8 @@ void WriteObjectWithKmsKey(google::cloud::storage::Client client, int& argc,
       stream << lineno << ": placeholder text for CMEK example.\n";
     }
 
-    gcs::ObjectMetadata meta = stream.Close();
+    stream.Close();
+    gcs::ObjectMetadata meta = stream.metadata().value();
     std::cout << "The resulting object size is: " << meta.size() << std::endl;
   }
   //! [write object with kms key] [END storage_upload_with_kms_key]

google/cloud/storage/internal/curl_client.cc

@@ -282,7 +282,7 @@ StatusOr<ResumableUploadResponse> CurlClient::UploadChunk(
   bool success_with_308 =
       response->status_code == 308 and
       response->headers.find("range") != response->headers.end();
-  if (status.ok() or success_with_308) {
+  if (response->status_code < 300 or success_with_308) {
     return ResumableUploadResponse::FromHttpResponse(*std::move(response));
   }
   return AsStatus(*response);
@@ -305,7 +305,7 @@ StatusOr<ResumableUploadResponse> CurlClient::QueryResumableUpload(
   bool success_with_308 =
       response->status_code == 308 and
       response->headers.find("range") != response->headers.end();
-  if (status.ok() or success_with_308) {
+  if (response->status_code < 300 or success_with_308) {
     return ResumableUploadResponse::FromHttpResponse(*std::move(response));
   }
   return AsStatus(*response);
@@ -1335,11 +1335,8 @@ StatusOr<ObjectMetadata> CurlClient::InsertObjectMediaMultipart(
   writer << crlf << request.contents() << crlf << marker << "--" << crlf;
 
   // 6. Return the results as usual.
-  auto response = writer.CloseRaw();
-  if (response.status_code >= 300) {
-    return Status(response.status_code, std::move(response.payload));
-  }
-  return ObjectMetadata::ParseFromString(response.payload);
+  writer.Close();
+  return std::move(writer).metadata();
 }
 
 std::string CurlClient::PickBoundary(std::string const& text_to_avoid) {

google/cloud/storage/internal/curl_resumable_streambuf.cc

@@ -42,8 +42,13 @@ void CurlResumableStreambuf::ValidateHash(ObjectMetadata const& meta) {
 }
 
 CurlResumableStreambuf::int_type CurlResumableStreambuf::overflow(int_type ch) {
-  Validate(__func__);
-  Flush(false);
+  if (not IsOpen()) {
+    return traits_type::eof();
+  }
+  auto status = Flush(false);
+  if (not status.ok()) {
+    return traits_type::eof();
+  }
   if (not traits_type::eq_int_type(ch, traits_type::eof())) {
     current_ios_buffer_.push_back(traits_type::to_char_type(ch));
     pbump(1);
@@ -52,34 +57,34 @@ CurlResumableStreambuf::int_type CurlResumableStreambuf::overflow(int_type ch) {
 }
 
 int CurlResumableStreambuf::sync() {
-  Flush(false);
+  auto status = Flush(false);
+  if (not status.ok()) {
+    return traits_type::eof();
+  }
   return 0;
 }
 
 std::streamsize CurlResumableStreambuf::xsputn(char const* s,
                                                std::streamsize count) {
-  Validate(__func__);
-  Flush(false);
+  if (not IsOpen()) {
+    return traits_type::eof();
+  }
+  auto status = Flush(false);
+  if (not status.ok()) {
+    return traits_type::eof();
+  }
+
   current_ios_buffer_.append(s, static_cast<std::size_t>(count));
   pbump(static_cast<int>(count));
   return count;
 }
 
-HttpResponse CurlResumableStreambuf::DoClose() {
+StatusOr<HttpResponse> CurlResumableStreambuf::DoClose() {
   GCP_LOG(INFO) << __func__ << "()";
   return Flush(true);
 }
 
-void CurlResumableStreambuf::Validate(char const* where) const {
-  if (IsOpen()) {
-    return;
-  }
-  std::string msg = "Attempting to use closed CurlResumableStreambuf in ";
-  msg += where;
-  google::cloud::internal::RaiseRuntimeError(msg);
-}
-
-HttpResponse CurlResumableStreambuf::Flush(bool final_chunk) {
+StatusOr<HttpResponse> CurlResumableStreambuf::Flush(bool final_chunk) {
   if (not IsOpen()) {
     return last_response_;
   }
@@ -105,6 +110,10 @@ HttpResponse CurlResumableStreambuf::Flush(bool final_chunk) {
   hash_validator_->Update(current_ios_buffer_);
 
   auto result = upload_session_->UploadChunk(current_ios_buffer_, upload_size);
+  if (not result.ok()) {
+    // This was an unrecoverable error, time to signal an error.
+    return std::move(result).status();
+  }
   current_ios_buffer_.clear();
   current_ios_buffer_.reserve(max_buffer_size_);
   setp(&current_ios_buffer_[0], &current_ios_buffer_[0] + max_buffer_size_);

google/cloud/storage/internal/curl_resumable_streambuf.h

@@ -57,15 +57,11 @@ class CurlResumableStreambuf : public ObjectWriteStreambuf {
   int sync() override;
   std::streamsize xsputn(char const* s, std::streamsize count) override;
   int_type overflow(int_type ch) override;
-  // TODO(coryan) this is an ugly return type.
-  HttpResponse DoClose() override;
+  StatusOr<HttpResponse> DoClose() override;
 
  private:
-  /// Raise an exception if the stream is closed.
-  void Validate(char const* where) const;
-
   /// Flush the libcurl buffer and swap it with the iostream buffer.
-  HttpResponse Flush(bool final_chunk);
+  StatusOr<HttpResponse> Flush(bool final_chunk);
 
   std::unique_ptr<ResumableUploadSession> upload_session_;
 

google/cloud/storage/internal/curl_streambuf.cc

@@ -159,37 +159,48 @@ std::streamsize CurlWriteStreambuf::xsputn(char const* s,
   return count;
 }
 
-HttpResponse CurlWriteStreambuf::DoClose() {
-  GCP_LOG(INFO) << __func__ << "()";
-  Validate(__func__);
-  SwapBuffers();
-  // TODO(#1735) - return StatusOr<> from here.
-  auto response = upload_.Close().value();
-  for (auto const& kv : response.headers) {
-    hash_validator_->ProcessHeader(kv.first, kv.second);
+StatusOr<HttpResponse> CurlWriteStreambuf::DoClose() {
+  GCP_LOG(DEBUG) << __func__ << "()";
+  Status status = Validate(__func__);
+  if (not status.ok()) {
+    return status;
+  }
+  status = SwapBuffers();
+  if (not status.ok()) {
+    return status;
+  }
+  auto response = upload_.Close();
+  if (response.ok()) {
+    for (auto const& kv : response->headers) {
+      hash_validator_->ProcessHeader(kv.first, kv.second);
+    }
   }
   return response;
 }
 
-void CurlWriteStreambuf::Validate(char const* where) const {
+Status CurlWriteStreambuf::Validate(char const* where) const {
   if (upload_.IsOpen()) {
-    return;
+    return Status();
   }
   std::string msg = "Attempting to use closed CurlStream in ";
   msg += where;
-  google::cloud::internal::RaiseRuntimeError(msg);
+  return Status(StatusCode::FAILED_PRECONDITION, std::move(msg));
 }
 
-void CurlWriteStreambuf::SwapBuffers() {
+Status CurlWriteStreambuf::SwapBuffers() {
   // Shorten the buffer to the actual used size.
   current_ios_buffer_.resize(pptr() - pbase());
   // Push the buffer to the libcurl wrapper to be written as needed
   hash_validator_->Update(current_ios_buffer_);
-  upload_.NextBuffer(current_ios_buffer_);
+  auto status = upload_.NextBuffer(current_ios_buffer_);
+  if (not status.ok()) {
+    return status;
+  }
   // Make the buffer big enough to receive more data before needing a flush.
   current_ios_buffer_.clear();
   current_ios_buffer_.reserve(max_buffer_size_);
   setp(&current_ios_buffer_[0], &current_ios_buffer_[0] + max_buffer_size_);
+  return Status();
 }
 
 }  // namespace internal

google/cloud/storage/internal/curl_streambuf.h

@@ -95,14 +95,14 @@ class CurlWriteStreambuf : public ObjectWriteStreambuf {
   int sync() override;
   std::streamsize xsputn(char const* s, std::streamsize count) override;
   int_type overflow(int_type ch) override;
-  HttpResponse DoClose() override;
+  StatusOr<HttpResponse> DoClose() override;
 
  private:
   /// Raise an exception if the stream is closed.
-  void Validate(char const* where) const;
+  Status Validate(char const* where) const;
 
   /// Flush the libcurl buffer and swap it with the iostream buffer.
-  void SwapBuffers();
+  Status SwapBuffers();
 
   CurlUploadRequest upload_;
   std::string current_ios_buffer_;

google/cloud/storage/internal/curl_upload_request.cc

@@ -36,26 +36,35 @@ CurlUploadRequest::CurlUploadRequest(std::size_t initial_buffer_size)
   buffer_rdptr_ = buffer_.end();
 }
 
-void CurlUploadRequest::Flush() {
+Status CurlUploadRequest::Flush() {
   ValidateOpen(__func__);
   handle_.FlushDebug(__func__);
   GCP_LOG(DEBUG) << __func__ << "(), curl.size=" << buffer_.size()
                  << ", curl.rdptr="
                  << std::distance(buffer_.begin(), buffer_rdptr_)
                  << ", curl.end="
                  << std::distance(buffer_.begin(), buffer_.end());
-  Wait([this] { return buffer_rdptr_ == buffer_.end(); });
+  return Wait([this] { return buffer_rdptr_ == buffer_.end(); });
 }
 
 StatusOr<HttpResponse> CurlUploadRequest::Close() {
-  ValidateOpen(__func__);
+  Status status = ValidateOpen(__func__);
+  if (not status.ok()) {
+    return status;
+  }
   handle_.FlushDebug(__func__);
-  Flush();
+  status = Flush();
+  if (not status.ok()) {
+    return status;
+  }
   // Set the the closing_ flag to trigger a return 0 from the next read
   // callback, see the comments in the header file for more details.
   closing_ = true;
   // Block until that callback is made.
-  Wait([this] { return curl_closed_; });
+  status = Wait([this] { return curl_closed_; });
+  if (not status.ok()) {
+    return status;
+  }
 
   // Now remove the handle from the CURLM* interface and wait for the response.
   auto error = curl_multi_remove_handle(multi_.get(), handle_.handle_.get());
@@ -71,11 +80,15 @@ StatusOr<HttpResponse> CurlUploadRequest::Close() {
                       std::move(received_headers_)};
 }
 
-void CurlUploadRequest::NextBuffer(std::string& next_buffer) {
-  ValidateOpen(__func__);
-  Flush();
+Status CurlUploadRequest::NextBuffer(std::string& next_buffer) {
+  Status status = ValidateOpen(__func__);
+  if (not status.ok()) {
+    return status;
+  }
+  status = Flush();
   next_buffer.swap(buffer_);
   buffer_rdptr_ = buffer_.begin();
+  return status;
 }
 
 Status CurlUploadRequest::SetOptions() {
@@ -216,13 +229,13 @@ Status CurlUploadRequest::AsStatus(CURLMcode result, char const* where) {
   return Status(StatusCode::UNKNOWN, std::move(os).str());
 }
 
-void CurlUploadRequest::ValidateOpen(char const* where) {
+Status CurlUploadRequest::ValidateOpen(char const* where) {
   if (not closing_) {
-    return;
+    return Status();
   }
-  std::ostringstream os;
-  os << "Attempting to use closed CurlUploadRequest in " << where;
-  google::cloud::internal::RaiseRuntimeError(os.str());
+  std::string msg = "Attempting to use closed CurlUploadRequest in ";
+  msg += where;
+  return Status(StatusCode::FAILED_PRECONDITION, std::move(msg));
 }
 
 }  // namespace internal