fix(storage): better error code for CreateBucket() and 409 errors (#10480)

It seems that 409 should be treated as a `kAlreadyExists` in this case,
and this case only.

Author: coryan

google/cloud/storage/internal/curl_client.cc

@@ -204,13 +204,20 @@ StatusOr<BucketMetadata> CurlClient::CreateBucket(
   // Assume the bucket name is validated by the caller.
   CurlRequestBuilder builder(storage_endpoint_ + "/b", storage_factory_);
   auto status = SetupBuilder(builder, request, "POST");
-  if (!status.ok()) {
-    return status;
-  }
+  if (!status.ok()) return status;
   builder.AddQueryParameter("project", request.project_id());
   builder.AddHeader("Content-Type: application/json");
-  return CheckedFromString<BucketMetadataParser>(
+  auto response = CheckedFromString<BucketMetadataParser>(
       std::move(builder).BuildRequest().MakeRequest(request.json_payload()));
+  // GCS returns a 409 when buckets already exist:
+  //     https://cloud.google.com/storage/docs/json_api/v1/status-codes#409-conflict
+  // This seems to be the only case where kAlreadyExists is a better match
+  // for 409 than kAborted.
+  if (!response && response.status().code() == StatusCode::kAborted) {
+    return Status(StatusCode::kAlreadyExists, response.status().message(),
+                  response.status().error_info());
+  }
+  return response;
 }
 
 StatusOr<BucketMetadata> CurlClient::GetBucketMetadata(

google/cloud/storage/internal/rest_client.cc

@@ -210,8 +210,18 @@ StatusOr<BucketMetadata> RestClient::CreateBucket(
   builder.AddQueryParameter("project", request.project_id());
   builder.AddHeader("Content-Type", "application/json");
   auto payload = request.json_payload();
-  return CheckedFromString<BucketMetadataParser>(storage_rest_client_->Post(
-      std::move(builder).BuildRequest(), {absl::MakeConstSpan(payload)}));
+  auto response =
+      CheckedFromString<BucketMetadataParser>(storage_rest_client_->Post(
+          std::move(builder).BuildRequest(), {absl::MakeConstSpan(payload)}));
+  // GCS returns a 409 when buckets already exist:
+  //     https://cloud.google.com/storage/docs/json_api/v1/status-codes#409-conflict
+  // This seems to be the only case where kAlreadyExists is a better match
+  // for 409 than kAborted.
+  if (!response && response.status().code() == StatusCode::kAborted) {
+    return Status(StatusCode::kAlreadyExists, response.status().message(),
+                  response.status().error_info());
+  }
+  return response;
 }
 
 StatusOr<BucketMetadata> RestClient::GetBucketMetadata(

google/cloud/storage/tests/bucket_integration_test.cc

@@ -33,6 +33,7 @@ namespace {
 using ::google::cloud::storage::testing::AclEntityNames;
 using ::google::cloud::testing_util::ContainsOnce;
 using ::google::cloud::testing_util::IsOk;
+using ::google::cloud::testing_util::StatusIs;
 using ::testing::Contains;
 using ::testing::HasSubstr;
 using ::testing::IsEmpty;
@@ -155,6 +156,23 @@ TEST_F(BucketIntegrationTest, BasicCRUD) {
   EXPECT_THAT(list_bucket_names(), Not(Contains(bucket_name)));
 }
 
+TEST_F(BucketIntegrationTest, CreateDuplicate) {
+  StatusOr<Client> client = MakeBucketIntegrationTestClient();
+  ASSERT_STATUS_OK(client);
+  auto const bucket_name = MakeRandomBucketName();
+  auto metadata = client->CreateBucketForProject(bucket_name, project_id_,
+                                                 BucketMetadata());
+  ASSERT_STATUS_OK(metadata);
+  ScheduleForDelete(*metadata);
+  EXPECT_EQ(bucket_name, metadata->name());
+  // Wait at least 2 seconds before trying to create / delete another bucket.
+  if (!UsingEmulator()) std::this_thread::sleep_for(std::chrono::seconds(2));
+
+  auto dup = client->CreateBucketForProject(bucket_name, project_id_,
+                                            BucketMetadata());
+  EXPECT_THAT(dup, StatusIs(StatusCode::kAlreadyExists));
+}
+
 TEST_F(BucketIntegrationTest, CreatePredefinedAcl) {
   std::vector<PredefinedAcl> test_values{
       PredefinedAcl::AuthenticatedRead(), PredefinedAcl::Private(),