Fix a bug in checking CopyObject response

Summary:
According to AWS docs:
```
A copy request might return an error when Amazon S3 receives the copy request or while Amazon S3 is copying the files. If the error occurs before the copy operation starts, you receive a standard Amazon S3 error. If the error occurs during the copy operation, the error response is embedded in the 200 OK response. This means that a 200 OK response can contain either a success or an error. Design your application to parse the contents of the response and handle it appropriately.
```
That means even if we receive 200, it still might fail.
According to another doc: https://sdk.amazonaws.com/cpp/api/LATEST/class_aws_1_1_s3_1_1_model_1_1_copy_object_result_details.html
```
The source and destination ETag is identical for a successfully copied object
```
That means we need to compare the 2 ETags to be sure.

Test Plan: unit test. also deploy in my namespace.

Reviewers: dhruba, igor

Reviewed By: dhruba

Differential Revision: https://rockset.phacility.com/D6325

Author: Duc Hieu Pham

cloud/aws/aws_s3.cc

@@ -52,6 +52,7 @@
 #include "util/string_util.h"
 
 namespace rocksdb {
+
 #ifdef USE_AWS
 class CloudRequestCallbackGuard {
  public:
@@ -159,12 +160,25 @@ class AwsS3ClientWrapper {
     return outcome;
   }
 
+  // The copy RPC is only successful iff outcome is success and etags between
+  // src and dst match.
+  static bool isCopyCloudObjectSuccess(
+      Aws::S3::Model::CopyObjectOutcome& outcome, const Aws::String& src_etag) {
+    bool success = false;
+    if (outcome.IsSuccess()) {
+      const auto& detail = outcome.GetResult().GetCopyObjectResultDetails();
+      success = (detail.ETagHasBeenSet() && detail.GetETag() == src_etag);
+    }
+    return success;
+  }
+
   Aws::S3::Model::CopyObjectOutcome CopyCloudObject(
-      const Aws::S3::Model::CopyObjectRequest& request) {
+      const Aws::S3::Model::CopyObjectRequest& request,
+      const Aws::String& src_etag) {
     CloudRequestCallbackGuard t(cloud_request_callback_.get(),
                                 CloudRequestOpType::kCopyOp);
     auto outcome = client_->CopyObject(request);
-    t.SetSuccess(outcome.IsSuccess());
+    t.SetSuccess(isCopyCloudObjectSuccess(outcome, src_etag));
     return outcome;
   }
 
@@ -746,6 +760,25 @@ Status S3StorageProvider::CopyCloudObject(const std::string& bucket_name_src,
 
   Aws::String src_url = src_bucket + src_object;
 
+  // Get the metadata of the source object. We need the metadata so that we can
+  // compare the ETag between the source object and dest object. Only when these
+  // ETags match can we conclude this CopyObject operations succeed.
+  //
+  // More details:
+  // https://sdk.amazonaws.com/cpp/api/LATEST/class_aws_1_1_s3_1_1_model_1_1_copy_object_result_details.html
+  // https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html
+  Aws::S3::Model::HeadObjectRequest head_request;
+  head_request.SetBucket(ToAwsString(bucket_name_src));
+  head_request.SetKey(ToAwsString(object_path_src));
+
+  auto head_outcome = s3client_->HeadObject(head_request);
+  if (!head_outcome.IsSuccess()) {
+    return Status::NotFound(
+        "[s3] CopyCloudObject: Fail to get metadata of src object: %s",
+        head_outcome.GetError().GetMessage().c_str());
+  }
+  const auto& src_etag = head_outcome.GetResult().GetETag();
+
   // create copy request
   Aws::S3::Model::CopyObjectRequest request;
   request.SetCopySource(src_url);
@@ -755,15 +788,29 @@ Status S3StorageProvider::CopyCloudObject(const std::string& bucket_name_src,
 
   // execute request
   Aws::S3::Model::CopyObjectOutcome outcome =
-      s3client_->CopyCloudObject(request);
-  bool isSuccess = outcome.IsSuccess();
+      s3client_->CopyCloudObject(request, src_etag);
+  bool isSuccess =
+      AwsS3ClientWrapper::isCopyCloudObjectSuccess(outcome, src_etag);
   if (!isSuccess) {
-    const Aws::Client::AWSError<Aws::S3::S3Errors>& error = outcome.GetError();
-    std::string errmsg(error.GetMessage().c_str());
+    // S3 CopyObject RPC can report errors in 2 ways:
+    // 1. outcome.IsSuccess() = false
+    // 2. outcome.IsSuccess() = true but the outcome's ETag doesn't match the
+    // source's ETag.
+    if (!outcome.IsSuccess()) {
+      const Aws::Client::AWSError<Aws::S3::S3Errors>& error =
+          outcome.GetError();
+      std::string errmsg(error.GetMessage().c_str());
+      Log(InfoLogLevel::ERROR_LEVEL, env_->info_log_,
+          "[s3] S3WritableFile src path %s error in copying to %s %s",
+          src_url.c_str(), dest_object.c_str(), errmsg.c_str());
+      return Status::IOError(dest_object.c_str(), errmsg.c_str());
+    }
+
     Log(InfoLogLevel::ERROR_LEVEL, env_->info_log_,
-        "[s3] S3WritableFile src path %s error in copying to %s %s",
-        src_url.c_str(), dest_object.c_str(), errmsg.c_str());
-    return Status::IOError(dest_object.c_str(), errmsg.c_str());
+        "[s3] S3WritableFile src path %s error in copying to %s: CopyObject "
+        "etags don't match",
+        src_url.c_str(), dest_object.c_str());
+    return Status::IOError("CopyObject S3 etags don't match");
   }
   Log(InfoLogLevel::INFO_LEVEL, env_->info_log_,
       "[s3] S3WritableFile src path %s copied to %s %s", src_url.c_str(),

cloud/db_cloud_test.cc

@@ -1,11 +1,11 @@
 // Copyright (c) 2017 Rockset
 
+#include <gtest/gtest.h>
+
 #ifndef ROCKSDB_LITE
 
 #ifdef USE_AWS
 
-#include "rocksdb/cloud/db_cloud.h"
-
 #include <algorithm>
 #include <chrono>
 #include <cinttypes>
@@ -18,6 +18,8 @@
 #include "file/filename.h"
 #include "logging/logging.h"
 #include "rocksdb/cloud/cloud_storage_provider.h"
+#include "rocksdb/cloud/db_cloud.h"
+#include "rocksdb/env.h"
 #include "rocksdb/options.h"
 #include "rocksdb/status.h"
 #include "rocksdb/table.h"
@@ -1419,6 +1421,47 @@ TEST_F(CloudTest, CheckpointToCloud) {
       checkpoint_bucket.GetBucketName(), checkpoint_bucket.GetObjectPath());
 }
 
+// Basic test to copy object within S3.
+TEST_F(CloudTest, CopyObjectTest) {
+  CreateAwsEnv();
+
+  // We need to open an empty DB in order for epoch to work.
+  OpenDB();
+
+  std::string content = "This is a test file";
+  std::string fname = dbname_ + "/100000.sst";
+  std::string dst_fname = dbname_ + "/200000.sst";
+
+  {
+    std::unique_ptr<WritableFile> writableFile;
+    aenv_->NewWritableFile(fname, &writableFile, EnvOptions());
+    writableFile->Append(content);
+    writableFile->Fsync();
+  }
+
+  Status st = aenv_->GetCloudEnvOptions().storage_provider->CopyCloudObject(
+      aenv_->GetSrcBucketName(), aenv_->RemapFilename(fname),
+      aenv_->GetSrcBucketName(), dst_fname);
+  ASSERT_OK(st);
+
+  {
+    std::unique_ptr<CloudStorageReadableFile> readableFile;
+    st = aenv_->GetCloudEnvOptions().storage_provider->NewCloudReadableFile(
+        aenv_->GetSrcBucketName(), dst_fname, &readableFile, EnvOptions());
+    ASSERT_OK(st);
+
+    char scratch[100];
+    Slice result;
+    st = dynamic_cast<SequentialFile*>(readableFile.get())
+             ->Read(100, &result, scratch);
+    ASSERT_OK(st);
+    ASSERT_EQ(19, result.size());
+    ASSERT_EQ(result, Slice(content));
+  }
+
+  CloseDB();
+}
+
 #ifdef AWS_DO_NOT_RUN
 //
 // Verify that we can cache data from S3 in persistent cache.